package io.kiki.stack.netty.protocol;

import io.netty.buffer.ByteBuf;
import io.netty.buffer.ByteBufAllocator;
import io.netty.buffer.CompositeByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.handler.codec.ByteToMessageDecoder;
import io.netty.handler.codec.DecoderException;
import io.netty.util.internal.RecyclableArrayList;
import io.netty.util.internal.StringUtil;
import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;

import java.util.ArrayList;
import java.util.List;

/**
 * This class mainly hack the {@link io.netty.handler.codec.ByteToMessageDecoder} to provide batch submission capability.
 * This can be used the same way as ByteToMessageDecoder except the case your following inbound handler may get a decoded msg,
 * which actually is an array list, then you can submit the list of msgs to an executor to process. For example
 * <pre>
 *   if (msg instanceof List) {
 *       processorManager.getDefaultExecutor().execute(new Runnable() {
 *           public void run() {
 *               // batch submit to an executor
 *               for (Object m : (List<?>) msg) {
 *                   RpcCommandHandler.this.process(ctx, m);
 *               }
 *           }
 *       });
 *   } else {
 *       process(ctx, msg);
 *   }
 * </pre>
 * You can check the method {@link AbstractBatchDecoder#channelRead(ChannelHandlerContext, Object)} ()}
 * to know the detail modification.
 */

@Slf4j
@Setter
@Getter
public abstract class AbstractBatchDecoder extends ChannelInboundHandlerAdapter {
    /**
     * Cumulate {@link ByteBuf}s by merge them into one {@link ByteBuf}'s, using memory copies.
     */
    public static final Cumulator MERGE_CUMULATOR = new Cumulator() {//
        @Override
        public ByteBuf cumulate(ByteBufAllocator byteBufAllocator, ByteBuf byteBuf, ByteBuf in) {
            ByteBuf buffer;
            if (byteBuf.writerIndex() > byteBuf.maxCapacity() - in.readableBytes() || byteBuf.refCnt() > 1) {
                // Expand cumulation (by replace it) when either there is not more room in the buffer
                // or if the refCnt is greater then 1 which may happen when the user use slice().retain() or
                // duplicate().retain().
                //
                // See:
                // - https://github.com/netty/netty/issues/2327
                // - https://github.com/netty/netty/issues/1764
                buffer = expandCumulation(byteBufAllocator, byteBuf, in.readableBytes());
            } else {
                buffer = byteBuf;
            }
            buffer.writeBytes(in);
            in.release();
            return buffer;
        }
    };

    /**
     * Cumulate {@link ByteBuf}s by add them to a {@link CompositeByteBuf} and so do no memory copy whenever possible.
     * Be aware that {@link CompositeByteBuf} use a more complex indexing implementation so depending on your use-case
     * and the decoder implementation this may be slower then just use the {@link #MERGE_CUMULATOR}.
     */
    public static final Cumulator COMPOSITE_CUMULATOR = new Cumulator() {
        @Override
        public ByteBuf cumulate(ByteBufAllocator byteBufAllocator, ByteBuf buffers, ByteBuf in) {
            ByteBuf byteBuf;
            if (buffers.refCnt() > 1) {
                // Expand cumulation (by replace it) when the refCnt is greater then 1 which may happen when the user
                // use slice().retain() or duplicate().retain().
                //
                // See:
                // - https://github.com/netty/netty/issues/2327
                // - https://github.com/netty/netty/issues/1764
                byteBuf = expandCumulation(byteBufAllocator, buffers, in.readableBytes());
                byteBuf.writeBytes(in);
                in.release();
            } else {
                CompositeByteBuf compositeByteBuf;
                if (buffers instanceof CompositeByteBuf) {
                    compositeByteBuf = (CompositeByteBuf) buffers;
                } else {
                    compositeByteBuf = byteBufAllocator.compositeBuffer();
                    int readable = buffers.readableBytes();//需要提前获取值
                    compositeByteBuf.addComponent(buffers).writerIndex(readable);
                }
                compositeByteBuf.addComponent(in).writerIndex(compositeByteBuf.writerIndex() + in.readableBytes());
                byteBuf = compositeByteBuf;
            }
            return byteBuf;
        }
    };

    ByteBuf buffers;
    private Cumulator cumulator = MERGE_CUMULATOR;
    private boolean singleDecode;
    private boolean decodeWasNull;
    private boolean first;
    private int discardAfterReads = 16;
    private int numReads;

    static ByteBuf expandCumulation(ByteBufAllocator byteBufAllocator, ByteBuf buffers, int readable) {
        ByteBuf oldByteBufs = buffers;
        buffers = byteBufAllocator.buffer(oldByteBufs.readableBytes() + readable);
        buffers.writeBytes(oldByteBufs);
        oldByteBufs.release();
        return buffers;
    }

    /**
     * If {@code true} then only one message is decoded on each
     * {@link #channelRead(ChannelHandlerContext, Object)} call.
     * <p>
     * Default is {@code false} as this has performance impacts.
     */
    public boolean isSingleDecode() {
        return singleDecode;
    }

    /**
     * If set then only one message is decoded on each {@link #channelRead(ChannelHandlerContext, Object)}
     * call. This may be useful if you need to do some protocol upgrade and want to make sure nothing is mixed up.
     * <p>
     * Default is {@code false} as this has performance impacts.
     */
    public void setSingleDecode(boolean singleDecode) {
        this.singleDecode = singleDecode;
    }

    /**
     * Set the {@link Cumulator} to use for cumulate the received {@link ByteBuf}s.
     */
    public void setCumulator(Cumulator cumulator) {
        if (cumulator == null) {
            throw new NullPointerException("cumulator");
        }
        this.cumulator = cumulator;
    }

    /**
     * Set the number of reads after which {@link ByteBuf#discardSomeReadBytes()} are called and so free up memory.
     * The default is {@code 16}.
     */
    public void setDiscardAfterReads(int discardAfterReads) {
        if (discardAfterReads <= 0) {
            throw new IllegalArgumentException("discardAfterReads must be > 0");
        }
        this.discardAfterReads = discardAfterReads;
    }

    /**
     * Returns the actual number of readable bytes in the internal cumulative
     * buffer of this decoder. You usually do not need to rely on this value
     * to write a decoder. Use it only when you must use it at your own risk.
     * This method is a shortcut to {@link #internalBuffer() internalBuffer().readableBytes()}.
     */
    protected int actualReadableBytes() {
        return internalBuffer().readableBytes();
    }

    /**
     * Returns the internal cumulative buffer of this decoder. You usually
     * do not need to access the internal buffer directly to write a decoder.
     * Use it only when you must use it at your own risk.
     */
    protected ByteBuf internalBuffer() {
        if (buffers != null) {
            return buffers;
        } else {
            return Unpooled.EMPTY_BUFFER;
        }
    }

    @Override
    public final void handlerRemoved(ChannelHandlerContext channelHandlerContext) throws Exception {
        ByteBuf internalBuffer = internalBuffer();
        int readable = internalBuffer.readableBytes();
        if (readable > 0) {
            ByteBuf byteBuf = internalBuffer.readBytes(readable);
            internalBuffer.release();
            channelHandlerContext.fireChannelRead(byteBuf);
        } else {
            internalBuffer.release();
        }
        buffers = null;
        numReads = 0;
        channelHandlerContext.fireChannelReadComplete();
        handlerRemoved0(channelHandlerContext);
    }

    /**
     * Gets called after the {@link ByteToMessageDecoder} was removed from the actual context and it doesn't handle
     * events anymore.
     */
    protected void handlerRemoved0(ChannelHandlerContext channelHandlerContext) throws Exception {
    }

    /**
     * This method has been modified to check the size of decoded msgs, which is represented by the
     * local variable {@code RecyclableArrayList out}. If has decoded more than one msg,
     * then construct an array list to submit all decoded msgs to the pipeline.
     */
    @Override
    public void channelRead(ChannelHandlerContext channelHandlerContext, Object msg) throws Exception {
        if (msg instanceof ByteBuf) {
            RecyclableArrayList recyclableArrayList = RecyclableArrayList.newInstance();
            try {
                ByteBuf byteBuf = (ByteBuf) msg;
                first = buffers == null;
                if (first) {
                    buffers = byteBuf;
                } else {
                    buffers = cumulator.cumulate(channelHandlerContext.alloc(), buffers, byteBuf);
                }
                callDecode(channelHandlerContext, buffers, recyclableArrayList);
            } catch (DecoderException e) {
                throw e;
            } catch (Throwable t) {
                throw new DecoderException(t);
            } finally {
                if (buffers != null && !buffers.isReadable()) {
                    numReads = 0;
                    buffers.release();
                    buffers = null;
                } else
                    if (++numReads >= discardAfterReads) {
                        // We did enough reads already try to discard some bytes so we not risk to see a OOME.
                        // See https://github.com/netty/netty/issues/4275
                        numReads = 0;
                        discardSomeReadBytes();
                    }

                int size = recyclableArrayList.size();
                if (size == 0) {
                    decodeWasNull = true;
                } else
                    if (size == 1) {
                        channelHandlerContext.fireChannelRead(recyclableArrayList.get(0));
                    } else {
                        ArrayList<Object> ret = new ArrayList<Object>(size);
                        for (int i = 0; i < size; i++) {
                            ret.add(recyclableArrayList.get(i));
                        }
                        channelHandlerContext.fireChannelRead(ret);
                    }

                recyclableArrayList.recycle();
            }
        } else {
            channelHandlerContext.fireChannelRead(msg);
        }
    }

    @Override
    public void channelReadComplete(ChannelHandlerContext channelHandlerContext) throws Exception {
        numReads = 0;
        discardSomeReadBytes();
        if (decodeWasNull) {
            decodeWasNull = false;
            if (!channelHandlerContext.channel().config().isAutoRead()) {
                channelHandlerContext.read();
            }
        }
        channelHandlerContext.fireChannelReadComplete();
    }

    protected final void discardSomeReadBytes() {
        if (buffers != null && !first && buffers.refCnt() == 1) {
            // discard some bytes if possible to make more room in the
            // buffer but only if the refCnt == 1  as otherwise the user may have
            // used slice().retain() or duplicate().retain().
            //
            // See:
            // - https://github.com/netty/netty/issues/2327
            // - https://github.com/netty/netty/issues/1764
            buffers.discardSomeReadBytes();
        }
    }

    @Override
    public void channelInactive(ChannelHandlerContext channelHandlerContext) throws Exception {
        RecyclableArrayList recyclableArrayList = RecyclableArrayList.newInstance();
        try {
            if (buffers != null) {
                callDecode(channelHandlerContext, buffers, recyclableArrayList);
                decodeLast(channelHandlerContext, buffers, recyclableArrayList);
            } else {
                decodeLast(channelHandlerContext, Unpooled.EMPTY_BUFFER, recyclableArrayList);
            }
        } catch (DecoderException e) {
            throw e;
        } catch (Exception e) {
            throw new DecoderException(e);
        } finally {
            try {
                if (buffers != null) {
                    buffers.release();
                    buffers = null;
                }
                int size = recyclableArrayList.size();
                for (int i = 0; i < size; i++) {
                    channelHandlerContext.fireChannelRead(recyclableArrayList.get(i));
                }
                if (size > 0) {
                    // Something was read, call fireChannelReadComplete()
                    channelHandlerContext.fireChannelReadComplete();
                }
                channelHandlerContext.fireChannelInactive();
            } finally {
                // recycle in all cases
                recyclableArrayList.recycle();
            }
        }
    }

    /**
     * Called once data should be decoded from the given {@link ByteBuf}. This method will call
     * {@link #decode(ChannelHandlerContext, ByteBuf, List)} as long as decoding should take place.
     *
     * @param channelHandlerContext the {@link ChannelHandlerContext} which this {@link ByteToMessageDecoder} belongs to
     * @param in                    the {@link ByteBuf} from which to read data
     * @param out                   the {@link List} to which decoded messages should be added
     */
    protected void callDecode(ChannelHandlerContext channelHandlerContext, ByteBuf in, List<Object> out) {
        try {
            while (in.isReadable()) {
                int outSize = out.size();
                int oldInputLength = in.readableBytes();
                decode(channelHandlerContext, in, out);

                // Check if this handler was removed before continuing the loop.
                // If it was removed, it is not safe to continue to operate on the buffer.
                //
                // See https://github.com/netty/netty/issues/1664
                if (channelHandlerContext.isRemoved()) {
                    break;
                }

                if (outSize == out.size()) {
                    if (oldInputLength == in.readableBytes()) {
                        break;
                    } else {
                        continue;
                    }
                }

                if (oldInputLength == in.readableBytes()) {
                    throw new DecoderException(StringUtil.simpleClassName(getClass()) + ".decode() did not read anything but decoded a message.");
                }

                if (isSingleDecode()) {
                    break;
                }
            }
        } catch (DecoderException e) {
            throw e;
        } catch (Throwable cause) {
            throw new DecoderException(cause);
        }
    }

    /**
     * Is called one last time when the {@link ChannelHandlerContext} goes in-active. Which means the
     * {@link #channelInactive(ChannelHandlerContext)} was triggered.
     * <p>
     * By default this will just call {@link #decode(ChannelHandlerContext, ByteBuf, List)} but sub-classes may
     * override this for some special cleanup operation.
     */
    protected void decodeLast(ChannelHandlerContext channelHandlerContext, ByteBuf in, List<Object> out) throws Exception {
        decode(channelHandlerContext, in, out);
    }

    /**
     * Decode the from one {@link ByteBuf} to an other. This method will be called till either the input
     * {@link ByteBuf} has nothing to read when return from this method or till nothing was read from the input
     * {@link ByteBuf}.
     *
     * @param channelHandlerContext the {@link ChannelHandlerContext} which this {@link ByteToMessageDecoder} belongs to
     * @param in                    the {@link ByteBuf} from which to read data
     * @param out                   the {@link List} to which decoded messages should be added
     * @throws Exception is thrown if an error accour
     */
    protected abstract void decode(ChannelHandlerContext channelHandlerContext, ByteBuf in, List<Object> out) throws Exception;

    /**
     * Cumulate {@link ByteBuf}s.
     */
    public interface Cumulator {

        ByteBuf cumulate(ByteBufAllocator alloc, ByteBuf byteBufs, ByteBuf in);
    }
}